/*-------------------------------------------------------------------------
 *
 * IDENTIFICATION
 *	  imagemanage.c
 *
 *-------------------------------------------------------------------------
 */
#include "walminer_decode.h"
#include "wm_utils.h"
#include "storage/bufpage.h"
#include "common/pg_lzcompress.h"


static bool get_block_image(XLogReaderState *record, uint8 block_id, char *page);
static void append_image(ImageStore *image, char* page);
static bool imageEqueal(ImageStore *image1, ImageStore *image2);
static void read_page(int index, char* page);

/*
 * 获取这个wal记录中，block_id序号块内存储的FPI,出参为page
 */
static bool
get_block_image(XLogReaderState *record, uint8 block_id, char *page)
{
	DecodedBkpBlock *bkpb;
	char	   *ptr;
	char		tmp[BLCKSZ];

	if (!record->blocks[block_id].in_use)
		return false;
	if (!record->blocks[block_id].has_image)
		return false;

	bkpb = &record->blocks[block_id];
	ptr = bkpb->bkp_image;

	if (bkpb->bimg_info & BKPIMAGE_IS_COMPRESSED)
	{
		/* If a backup block image is compressed, decompress it */
#if (defined PG_VERSION_12)  || (defined PG_VERSION_13)
		if (pglz_decompress(ptr, bkpb->bimg_len, tmp,
							BLCKSZ - bkpb->hole_length,true) < 0)
#else
		if (pglz_decompress(ptr, bkpb->bimg_len, tmp,
							BLCKSZ - bkpb->hole_length) < 0)
#endif
		{
			elog(LOG, "invalid compressed image at %X/%X, block %d",
								  (uint32) (record->ReadRecPtr >> 32),
								  (uint32) record->ReadRecPtr,
								  block_id);
			return false;
		}
		ptr = tmp;
	}

	/* generate page, taking into account hole if necessary */
	if (bkpb->hole_length == 0)
	{
		memcpy(page, ptr, BLCKSZ);
	}
	else
	{
		memcpy(page, ptr, bkpb->hole_offset);
		/* must zero-fill the hole */
		MemSet(page + bkpb->hole_offset, 0, bkpb->hole_length);
		memcpy(page + (bkpb->hole_offset + bkpb->hole_length),
			   ptr + bkpb->hole_offset,
			   BLCKSZ - (bkpb->hole_offset + bkpb->hole_length));
	}

	return true;
}

void
clean_image(void)
{
	char	storefile[MAXPGPATH] = {0};
	char	path[MAXPGPATH] = {0};
	char	*filename = "storeimage";
	FILE	*fp = NULL;

	get_image_path((char*)path);
	sprintf(storefile,"%s/%s", path, filename);
	fp = fopen(storefile, "w");
	if(!fp)
	{
		elog(ERROR,"fail to open file %s to read", storefile);
	}
	fclose(fp);
}


static void
read_page(int index, char* page)
{
	char	storefile[MAXPGPATH] = {0};
	char	path[MAXPGPATH] = {0};
	char	*filename = "storeimage";
	FILE	*fp = NULL;
	int64	seeksize = 0;
	Assert(page);
	
	get_image_path((char*)path);
	sprintf(storefile,"%s/%s", path, filename);

	fp = fopen(storefile, "rb");
	if(!fp)
	{
		elog(ERROR,"fail to open file %s to read", storefile);
	}
	fseek(fp, 0, SEEK_SET);
	Assert(0 == ftell(fp));
	Assert(0 <= index);
	if(0 != index)
	{
		seeksize = (int64)index * BLCKSZ;
		fseek(fp, seeksize, SEEK_SET);
	}
	if(BLCKSZ != fread(page, 1, BLCKSZ, fp))
	{
		elog(ERROR,"fail to read %s", storefile);
	}

	fclose(fp);
	fp = NULL;
}

/*
 * 向硬盘存储FPW的代码
 * TODO(lchch)设计一个FPW的存盘机制，使解析效率增加
 */
void
flush_page(int index, char* page)
{
	char	storefile[MAXPGPATH] = {0};
	char	path[MAXPGPATH] = {0};
	char	*filename = "storeimage";
	FILE	*fp = NULL;
	int64	seeksize = 0;

	Assert(page);

    get_image_path((char*)path);
	sprintf(storefile,"%s/%s", path, filename);

	fp = fopen(storefile, "rb+");
	if(!fp)
	{
		elog(ERROR,"fail to open file %s to write:%m", storefile);
	}	
	fseek(fp, 0, SEEK_SET);
	Assert(0 == ftell(fp));
	Assert(0 <= index);
	if(0 != index)
	{
		seeksize = (int64)index * BLCKSZ;
		fseek(fp, seeksize, SEEK_SET);
		//walminer_debug("[flush_page]index=%d, ftell=%d,seeksize=%d",index, ftell(fp), seeksize);
		
	}
	if(BLCKSZ != fwrite(page, 1, BLCKSZ, fp))
	{
		elog(ERROR,"fail to write to %s", storefile);
	}
	fclose(fp);
	fp = NULL;
}

static void
append_image(ImageStore *image, char* page)
{
	int			listlength = 0;
	int			loop = 0;
	ImageStore 	*imagePtr = NULL;
	ListCell	*lc = NULL;
    List        *image_list = NULL;
	static int	imageondisklast = 0;
	int			reporstep = (512 * 1024 * 1024) / BLCKSZ;	//512M image on disk

    image_list = wdecoder.wdc->anapro.image_list;
    listlength = list_length(image_list);

	for(; loop < listlength; loop++)
	{
#ifdef PG_VERSION_13
		lc = lc?(lnext(image_list, lc)):(list_head(image_list));
#else
		lc = lc?(lnext(lc)):(list_head(image_list));
#endif
		imagePtr = (ImageStore*)lfirst(lc);
		if(imageEqueal(image,imagePtr))
		{
			pfree(image);
			flush_page(loop, page);
			return;
		}
	}
	flush_page(loop, page);

	wdecoder.wdc->anapro.image_list = lappend(image_list, image);
    image_list =  wdecoder.wdc->anapro.image_list;
	if(imageondisklast + reporstep <= image_list->length)
	{
		imageondisklast = image_list->length;
		elog(NOTICE, "there be %d image pages on disk", imageondisklast);
	}
}

void
record_store_image(XLogReaderState *record)
{
	uint8			block_id = 0;
	DecodedBkpBlock *bkpb = NULL;
	ImageStore 		*image = NULL;
    RelFileNode     target_node;
	char			page[BLCKSZ] = {0};

    memset(&target_node, 0, sizeof(RelFileNode));
    XLogRecGetBlockTag(record, 0, &target_node, NULL, NULL);
    if(!filter_by_relfilenode(&target_node))
	{
        return;
	}
	for(block_id = 0; block_id <= XLR_MAX_BLOCK_ID; block_id++)
	{
		bkpb = &record->blocks[block_id];
		if(!bkpb->in_use)
			return;
		if(!bkpb->has_image)
			continue;
		memset(page, 0, BLCKSZ);
		image = palloc0(sizeof(ImageStore));
		image->rnode.dbNode = bkpb->rnode.dbNode;
		image->rnode.relNode = bkpb->rnode.relNode;
		image->rnode.spcNode = bkpb->rnode.spcNode;

		image->forknum = bkpb->forknum;
		image->blkno = bkpb->blkno;


		if(get_block_image(record, block_id, page))
		{
			walminer_debug("record_store_image relNode=%d, blkno=%d", image->rnode.relNode, image->blkno );
			//out_page_to_file(page, image->rnode.relNode, image->blkno, 0);
			append_image(image, page);
		}
		else
		{
			walminer_debug("record_store_image NULL");
		}
	}
}

void
page_init_by_xlog(RelFileNode *rnode, ForkNumber forknum, BlockNumber blkno)
{
	ImageStore 		*image = NULL;
	char 			page[BLCKSZ] = {0};
	
	PageInit(page, BLCKSZ, 0);

	image = palloc(sizeof(ImageStore));
	image->rnode.dbNode = rnode->dbNode;
	image->rnode.relNode = rnode->relNode;
	image->rnode.spcNode = rnode->spcNode;
	if(debug_mode)
	{
		walminer_debug("[page_init_by_xlog]initpage:relnode %u, blckno %d",rnode->relNode, blkno);
	}

	image->forknum = forknum;
	image->blkno = blkno;
	append_image(image, page);
}

bool
get_image_from_store(RelFileNode *rnode, ForkNumber forknum, BlockNumber blkno, char* page, int *index)
{
	ImageStore	images;
	int			loop = 0;
	int			listlength = 0;
	ListCell	*lc = NULL;
    List           *image_list = NULL;
	ImageStore	*imagesPtr = NULL, *imagesTar = NULL;

	Assert(rnode);
	Assert(page);
	memset(&images, 0, sizeof(ImageStore));

	memcpy(&images.rnode, rnode, sizeof(RelFileNode));
	images.forknum = forknum;
	images.blkno = blkno;

    

    image_list = wdecoder.wdc->anapro.image_list;
	listlength = list_length(image_list);

    if(imageEqueal(&images, wdecoder.wdc->anapro.current_image))
    {
        imagesTar = wdecoder.wdc->anapro.current_image;
        loop =  wdecoder.wdc->anapro.current_image_index;
    }
    else
    {
        for(; loop < listlength; loop++)
        {
#ifdef PG_VERSION_13
			lc = lc?(lnext(image_list, lc)):(list_head(image_list));
#else
			lc = lc?(lnext(lc)):(list_head(image_list));
#endif
            imagesPtr = (ImageStore*)lfirst(lc);
            if(imageEqueal(&images, imagesPtr))
            {
                imagesTar = imagesPtr;
                break;
            }
        }
    }

	if(!imagesTar)
	{
        walminer_debug("[getImageFromStore]page:relnode %u, blckno %d NOT FOUND", rnode->relNode, blkno);
		return false;
	}

	read_page(loop, page);
	//out_page_to_file(page, images.rnode.relNode, images.blkno, 1);
	*index = loop;
	walminer_debug("[getImageFromStore]page:relnode %u, blckno %d,index %d", rnode->relNode, blkno, *index);
	return true;
}

void
page_init_in_wal(RelFileNode *rnode, ForkNumber forknum, BlockNumber blkno)
{
	ImageStore 		*image = NULL;
	char 			page[BLCKSZ] = {0};
	
	PageInit(page, BLCKSZ, 0);

	image = (ImageStore*)walminer_malloc(sizeof(ImageStore), 0);
	image->rnode.dbNode = rnode->dbNode;
	image->rnode.relNode = rnode->relNode;
	image->rnode.spcNode = rnode->spcNode;

	walminer_debug("[page_init_in_wal]initpage:relnode %u, blckno %d",rnode->relNode, blkno);

	image->forknum = forknum;
	image->blkno = blkno;
	append_image(image, page);
}

static bool
imageEqueal(ImageStore *image1, ImageStore *image2)
{
	if(!image2)
		return false;

	if(image1->forknum != image2->forknum || image1->blkno != image2->blkno)
		return false;
	if(0 != memcmp(&image1->rnode, &image2->rnode, sizeof(RelFileNode)))
		return false;

	return true;
}

void
out_page_to_file(char *page, Oid relfilenode, BlockNumber blkno, int kind)
{
	char	log_path[MAXPGPATH] = {0};
	char	file_path[MAXPGPATH] = {0};
	FILE	*fp = NULL;

	get_runtemp_path(log_path);
	sprintf(file_path, "%s/%u_%u_%d", log_path, blkno, relfilenode, kind);
	
	fp = fopen(file_path, "wb");
	if(!fp)
	{
		elog(ERROR, "can not open %s to write:%m", file_path);
	}
	fwrite(page, 1, BLCKSZ, fp);
	fclose(fp);
}
